package com.hujiang.basic.search.zookeeper.zkLock;

import com.hujiang.basic.search.handler.HandlerService;
import com.hujiang.basic.search.handler.impl.model.HandlerInfo;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created by zhanglibin on 2017/3/22.
 */
@Slf4j
@Component
public class DistributedLock {

    public Map<String, InterProcessMutex> fullIndexLocks = new ConcurrentHashMap<>();
    public InterProcessMutex incLock;

    @Value("${zk.lock.prefix}")
    private String prefixLock;

    @Value(value = "${zk.url}")
    private String zkServers;

    @Value(value = "${increment.index.lock}")
    private String incLockPath;

    @Autowired
    private HandlerService service;

    @PostConstruct
    private void initIncLock() {
        incLock = new InterProcessMutex(getCuratorClient(), incLockPath);
        log.info("init IncLock success");
    }

    @PostConstruct
    @Scheduled(fixedRate = 60_000)
    private void initDistributedLock() {

        List<HandlerInfo> queues = service.query();
        if (CollectionUtils.isEmpty(queues)) {
            log.error("no queue when init zookeeper client");
            return;
        }
        queues.forEach((q) -> {
            if (StringUtils.isEmpty(q.getAlias())) return;
            if (fullIndexLocks.containsKey(q.getAlias())) return;
            try {
                InterProcessMutex lock = new InterProcessMutex(getCuratorClient(), prefixLock + q.getAlias());
                fullIndexLocks.put(q.getAlias(), lock);
                log.info("init fullIndexLocks :{} success", q.getAlias());
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        });
    }


    private CuratorFramework getCuratorClient() {
        CuratorFramework curator = CuratorFrameworkFactory.builder().retryPolicy(
                new ExponentialBackoffRetry(10000, 3)).connectString(zkServers).build();
        curator.getConnectionStateListenable().addListener(
                new ConnectionStateListener() {
                    @Override
                    public void stateChanged(CuratorFramework client, ConnectionState newState) {
                        log.info("current zookeeper connection status : {}", newState);
                        if (newState == ConnectionState.LOST) {//处理session过期
                            log.info("connection lost");
                        }
                    }
                });
        curator.start();
        return curator;
    }

}
